package avicit.bdp.dcs.semistructure.service;

import avicit.bdp.common.dto.BdpFileStatus;
import avicit.bdp.common.dto.StorageResourceConf;
import avicit.bdp.core.constant.Constants;
import avicit.bdp.core.util.json.JSONUtils;
import avicit.bdp.common.utils.CheckParaUtils;
import avicit.bdp.common.utils.uploads.CatalogBean;
import avicit.bdp.common.utils.uploads.FileAdapterUtils;
import avicit.bdp.common.utils.uploads.IFileAdapter;
import avicit.bdp.dcs.datasource.service.DataSourceUtils;
import avicit.bdp.dcs.job.dto.JobDTO;
import avicit.bdp.dcs.job.service.DcsJobService;
import avicit.bdp.dcs.semistructure.dto.FileFilterRuleDto;
import avicit.bdp.dcs.semistructure.dto.JobFileDTO;
import avicit.bdp.dcs.task.dto.TaskDTO;
import avicit.bdp.dcs.tools.enums.Enable;
import avicit.bdp.dcs.tools.enums.SyncMode;
import avicit.bdp.dcs.utils.DcsErrorCode;
import com.alibaba.fastjson2.JSONObject;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.*;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

/**
 * @金航数码科技有限责任公司
 * @作者：liyb
 * @邮箱：liyb@avic-digital.com
 * @创建时间： 2023-04-10 10:53
 * @类说明： 文件管理实现类
 * @修改记录：
 */
@Service
@Transactional
public class FileManageService {

  private static final Logger logger = LoggerFactory.getLogger(FileManageService.class);

  @Autowired
  private DcsJobService jobService;
  @Autowired
  private DataSourceUtils dataSourceUtils;
  @Autowired
  private JobFileService jobFileService;

  /**
   * 查询目录和文件
   * @param catalogPath
   * @param jobId
   * @return
   * @throws Exception
   */
  public CatalogBean getCatalogAndFileTree(String catalogPath, String jobId) throws Exception {
    JobDTO jobDTO = jobService.selectByPrimaryKey(jobId);
    CheckParaUtils.checkDbObjectValidity(jobDTO, logger, String.format(DcsErrorCode.JOB_NULL.getDesc(), jobId));

    String srcDatasourceId = jobDTO.getSrcDatasourceId();
    String systemBuilt = jobDTO.getExtraParams();
    String projectId = jobDTO.getProjectId();

    IFileAdapter fileAdapter = null;
    // 系统内置数据源
    if(Enable.ONE.getDescp().equals(systemBuilt)){
      fileAdapter = FileAdapterUtils.getFileAdapterByProject(projectId);
    }else{
      // 数据源中的数据源
      StorageResourceConf conf = dataSourceUtils.getStorageResourceByDatasourceId(srcDatasourceId);
      fileAdapter = FileAdapterUtils.getFileService(conf);
    }

    return fileAdapter.getFileTreeList(catalogPath, "file", "");
  }

  public List<CatalogBean> getCatalogAndFileTreeByCatalogPath(String catalogPath, String jobId) throws Exception {
    CatalogBean catalogBean = getCatalogAndFileTree(catalogPath, jobId);
    List<CatalogBean> childrenList = catalogBean.getChildren();
    return childrenList;
  }

  /**
   * 获取待同步的文件
   * @return
   * @throws Exception
   */
  public String getSyncFile(TaskDTO taskDTO, JobDTO jobDTO) throws Exception {
    String selectedFile = jobDTO.getSelectedFile();
    List<FileFilterRuleDto> fileFilterRuleList = JSONUtils.toList(selectedFile, FileFilterRuleDto.class);
    FileFilterRuleDto fileFilter = null;
    if (CollectionUtils.isNotEmpty(fileFilterRuleList)) {
      for(FileFilterRuleDto fileFilterRule : fileFilterRuleList){
        if(fileFilterRule.getTemplateId().equals(taskDTO.getSrcTableName())){
          fileFilter = fileFilterRule;
          break;
        }
      }
    }

    // 获取已经同步过的文件
//    List<JobFileDTO> jobFileList = getJobFileList(taskDTO.getId());
    List<JobFileDTO> jobFileList = new ArrayList<>();

    return getSelectCatalogFile(fileFilter, taskDTO, jobDTO, jobFileList);
  }

  /**
   * 获取已经同步过的文件
   * @param taskId
   * @return
   */
  private List<JobFileDTO> getJobFileList(String taskId){
    JobFileDTO jobFileDTOQuery = new JobFileDTO();
    jobFileDTOQuery.setTaskId(taskId);
    return jobFileService.selectList(jobFileDTOQuery);
  }

  /**
   * 获取选中目录中的文件
   * @param jobDTO
   * @param jobFileList
   * @return
   * @throws Exception
   */
  private String getSelectCatalogFile(FileFilterRuleDto fileFilterRule, TaskDTO taskDTO,
                                      JobDTO jobDTO, List<JobFileDTO> jobFileList) throws Exception {
    String srcDatasourceId = jobDTO.getSrcDatasourceId();
    String systemBuilt = jobDTO.getExtraParams();
    String projectId = jobDTO.getProjectId();
    boolean containSubdirectory = fileFilterRule.isContainSubdirectory();

    IFileAdapter fileAdapter = null;
    // 系统内置数据源
    if(Enable.ONE.getDescp().equals(systemBuilt)){
      fileAdapter = FileAdapterUtils.getFileAdapterByProject(projectId);
    }else{
      // 数据源中的数据源
      StorageResourceConf conf = dataSourceUtils.getStorageResourceByDatasourceId(srcDatasourceId);
      fileAdapter = FileAdapterUtils.getFileService(conf);
    }

    Map<String, Long> fileDateMap = new HashMap<>(64);
    List<String> catalogList = new ArrayList<>();
    List<String> fileList = new ArrayList<>();
    List<CatalogBean> catalogBeanList = fileFilterRule.getCatalogBeanList();
    if (CollectionUtils.isNotEmpty(catalogBeanList)) {
      for(CatalogBean catalogBean : catalogBeanList){
        if(catalogBean.isDirectory()){
          catalogList.add(catalogBean.getCatalogPath());
        }else{
          fileList.add(catalogBean.getCatalogPath());

          BdpFileStatus bdpFileStatus = fileAdapter.getFileStatus(catalogBean.getCatalogPath());
          if(bdpFileStatus != null){
            fileDateMap.put(catalogBean.getCatalogPath(), bdpFileStatus.getModificationTime());
          }
        }
      }
    }

    // 获取所有的文件
    List<String> allFileList = new ArrayList<>();
    allFileList.addAll(fileList);
    logger.info("选择的文件allFileList=" + allFileList);

    for(String catalogPath : catalogList){
      if(containSubdirectory){
        // 递归查找子目录文件
        allFileList.addAll(getSubdirectoryFileList(fileAdapter, catalogPath, fileDateMap));
      }else{
        CatalogBean catalogBean = fileAdapter.getFileTreeList(catalogPath, "file", "");
        // 查找指定目录下的文件
        allFileList.addAll(getFileListByCatalogPath(catalogBean, fileDateMap));
      }
    }

    logger.info("加上目录中的文件allFileList=" + allFileList);

    // 文件过滤规则,过滤文件
    allFileList = filterFile(allFileList, fileFilterRule, fileDateMap);

    logger.info("过滤后文件allFileList=" + allFileList);

    // 增量采集按文件时间过滤
    logger.info("fileDateMap=" + fileDateMap);
    allFileList = filterByFileDate(allFileList, taskDTO, fileDateMap);
    logger.info("增量采集按文件时间过滤后文件allFileList=" + allFileList);

    // 排除已经同步的文件
    return filterBySyncFile(allFileList, jobFileList);
  }

  /**
   * 查找指定目录下的文件
   * @param catalogBean
   * @return
   */
  private List<String> getFileListByCatalogPath(CatalogBean catalogBean, Map<String, Long> fileDateMap){
    List<String> fileList = new ArrayList<>();
    List<CatalogBean> childrenList = catalogBean.getChildren();
    if (CollectionUtils.isNotEmpty(childrenList)) {
      for(CatalogBean catalog : childrenList){
        if(!catalog.isDirectory()){
          fileList.add(catalog.getCatalogPath());
          fileDateMap.put(catalog.getCatalogPath(), catalog.getModificationTime());
        }
      }
    }

    return fileList;
  }

  /**
   * 文件过滤规则,过滤文件
   * @param fileList
   * @return
   */
  private List<String> filterFile(List<String> fileList, FileFilterRuleDto fileFilterRule, Map<String, Long> fileDateMap){
    String fileType = fileFilterRule.getFileType();
    String fileNameStart = fileFilterRule.getFileNameStart();
    String fileNameEnd = fileFilterRule.getFileNameEnd();
    String fileNameContain = fileFilterRule.getFileNameContain();
    String fileNameCron = fileFilterRule.getFileNameCron();

    List<String> filterFileList = new ArrayList<>();
    filterFileList = filterByType(fileList, fileType, fileDateMap);
    filterFileList = filterByFileNameStart(filterFileList, fileNameStart, fileDateMap);
    filterFileList = filterByFileNameEnd(filterFileList, fileNameEnd, fileDateMap);
    filterFileList = filterByFileNameContain(filterFileList, fileNameContain, fileDateMap);
    filterFileList = filterByFileNameCron(filterFileList, fileNameCron, fileDateMap);

    return filterFileList;
  }

  /**
   * 按文件类型过滤
   * @param fileList
   * @param fileType
   * @return
   */
  private List<String> filterByType(List<String> fileList, String fileType, Map<String, Long> fileDateMap){
    List<String> filterFileList = new ArrayList<>();
    if(StringUtils.isNotEmpty(fileType)){
      String[] arr = fileType.trim().split(Constants.COMMA);
      List<String> fileTypeList = new ArrayList<>();
      for(String fileType_ : arr){
        fileTypeList.add(fileType_);
      }

      for(String fileName : fileList){
        if(fileTypeList.contains(fileName.substring(fileName.lastIndexOf(".") + 1))){
          filterFileList.add(fileName);
        }else{
          fileDateMap.remove(fileName);
        }
      }
      return filterFileList;
    }else{
      return fileList;
    }
  }

  /**
   * 按文件名开头过滤
   * @param fileList
   * @param fileNameStart
   * @return
   */
  private List<String> filterByFileNameStart(List<String> fileList, String fileNameStart, Map<String, Long> fileDateMap){
    List<String> filterFileList = new ArrayList<>();
    if(StringUtils.isNotEmpty(fileNameStart)){
      String[] arr = fileNameStart.trim().split(Constants.COMMA);
      List<String> fileNameStartList = new ArrayList<>();
      for(String fileNameStart_ : arr){
        fileNameStartList.add(fileNameStart_);
      }

      for(String fileName : fileList){
        String shortFileName = fileName.substring(fileName.lastIndexOf("/") + 1);
        boolean flag = false;
        for(String pattern : fileNameStartList){
          if(shortFileName.startsWith(pattern)){
            flag = true;
            break;
          }
        }

        if(flag){
          filterFileList.add(fileName);
        }else{
          fileDateMap.remove(fileName);
        }
      }
      return filterFileList;
    }else{
      return fileList;
    }
  }

  /**
   * 按文件名结尾过滤
   * @param fileList
   * @param fileNameEnd
   * @return
   */
  private List<String> filterByFileNameEnd(List<String> fileList, String fileNameEnd, Map<String, Long> fileDateMap){
    List<String> filterFileList = new ArrayList<>();
    if(StringUtils.isNotEmpty(fileNameEnd)){
      String[] arr = fileNameEnd.trim().split(Constants.COMMA);
      List<String> fileNameEndList = new ArrayList<>();
      for(String fileNameEnd_ : arr){
        fileNameEndList.add(fileNameEnd_);
      }

      for(String fileName : fileList){
        boolean flag = false;
        for(String pattern : fileNameEndList){
          if(fileName.substring(0, fileName.lastIndexOf(".")).endsWith(pattern)){
            flag = true;
            break;
          }
        }

        if(flag){
          filterFileList.add(fileName);
        }else{
          fileDateMap.remove(fileName);
        }
      }
      return filterFileList;
    }else{
      return fileList;
    }
  }

  /**
   * 按文件名包含过滤
   * @param fileList
   * @param fileNameContain
   * @return
   */
  private List<String> filterByFileNameContain(List<String> fileList, String fileNameContain, Map<String, Long> fileDateMap){
    List<String> filterFileList = new ArrayList<>();
    if(StringUtils.isNotEmpty(fileNameContain)){
      String[] arr = fileNameContain.trim().split(Constants.COMMA);
      List<String> fileNameContainList = new ArrayList<>();
      for(String fileNameContain_ : arr){
        fileNameContainList.add(fileNameContain_);
      }

      for(String fileName : fileList){
        boolean flag = false;
        for(String pattern : fileNameContainList){
          if(fileName.substring(0, fileName.lastIndexOf(".")).contains(pattern)){
            flag = true;
            break;
          }
        }

        if(flag){
          filterFileList.add(fileName);
        }else{
          fileDateMap.remove(fileName);
        }
      }
      return filterFileList;
    }else{
      return fileList;
    }
  }

  /**
   * 按文件名正则过滤
   * @param fileList
   * @param fileNameCron
   * @return
   */
  private List<String> filterByFileNameCron(List<String> fileList, String fileNameCron, Map<String, Long> fileDateMap){
    List<String> filterFileList = new ArrayList<>();
    if(StringUtils.isNotEmpty(fileNameCron)){
      Pattern pattern = Pattern.compile(fileNameCron.trim());

      for(String fileName : fileList){
        String shortFileName = fileName.substring(fileName.lastIndexOf("/") + 1);
        if(pattern.matcher(shortFileName).matches()){
          filterFileList.add(fileName);
        }else{
          fileDateMap.remove(fileName);
        }
      }
      return filterFileList;
    }else{
      return fileList;
    }
  }

  /**
   * 过滤已经同步的文件
   * @param fileList
   * @param jobFileList
   * @return
   */
  private String filterBySyncFile(List<String> fileList, List<JobFileDTO> jobFileList){
    String hdfsFilePath = "";

    // 排除已经同步的文件
    if (CollectionUtils.isNotEmpty(jobFileList)) {
      List<String> resultList = new ArrayList<>();
      for (String filePath : fileList) {
        boolean flag = true;
        for (JobFileDTO jobFile : jobFileList) {
          if (jobFile.getFilePath().contains(filePath)) {
            flag = false;
            break;
          }
        }

        if (flag) {
          resultList.add(filePath);
        }
      }

      hdfsFilePath = String.join(Constants.COMMA, resultList);
    } else {
      logger.info("待同步的文件个数：{}", fileList.size());
      if (fileList.size() > 0) {
        hdfsFilePath = String.join(Constants.COMMA, fileList);
      }
    }

    return hdfsFilePath;
  }

  /**
   * 递归查找子目录文件
   * @param catalogPath
   * @return
   */
  private List<String> getSubdirectoryFileList(IFileAdapter fileAdapter, String catalogPath, Map<String, Long> fileDateMap) throws Exception{
    List<String> fileList = new ArrayList<>();

    CatalogBean catalogBean = fileAdapter.getFileTreeList(catalogPath, "file", "");
    // 查找指定目录下的文件
    fileList.addAll(getFileListByCatalogPath(catalogBean, fileDateMap));

    String subCatalog = null;
    List<CatalogBean> childrenList = catalogBean.getChildren();
    if (CollectionUtils.isNotEmpty(childrenList)) {
      for(CatalogBean catalog : childrenList){
        if(catalog.isDirectory()){
          subCatalog = catalog.getCatalogPath();
          fileList.addAll(getSubdirectoryFileList(fileAdapter, subCatalog, fileDateMap));
        }
      }
    }

    return fileList;
  }

  /**
   * 增量采集按文件时间过滤
   * @param fileList
   * @param taskDTO
   * @return
   */
  private List<String> filterByFileDate(List<String> fileList, TaskDTO taskDTO, Map<String, Long> fileDateMap) throws Exception{
    List<String> filterFileList = new ArrayList<>();
    if(taskDTO.getSyncMode() == SyncMode.DATAX_INCREMENT_SYNC.getCode()){
      return getFileListByTimeSort(fileDateMap, taskDTO.getSyncIdValue());
    }else{
      return fileList;
    }
  }

  /**
   * 按时间升序返回文件列表
   * @param map
   * @return
   */
  private List<String> getFileListByTimeSort(Map<String, Long> map, Long syncIdValue) throws Exception{
    List<String> fileList = new ArrayList<>();
    if (map == null) {
      return fileList;
    }

    long fileModifyTime = 0L;
    if(syncIdValue != null){
      fileModifyTime = syncIdValue.longValue();
    }

    Collection<Long> c = map.values();
    Object[] objArr = c.toArray();
    Arrays.sort(objArr);

    String key = null;
    long modificationTime;
    for(Object obj : objArr){
      modificationTime = Long.parseLong(obj.toString());
      if(fileModifyTime != 0 && modificationTime <= fileModifyTime){
        continue;
      }

      key = getKey(map, modificationTime);
      if(StringUtils.isNotEmpty(key)){
        fileList.add(key);
      }
    }
    return fileList;
  }

  private String getKey(Map<String, Long> map, long v) {
    Iterator<String> it = map.keySet().iterator();
    while (it.hasNext()) {
      String key = it.next();
      long value = map.get(key);
      if (value == v) {
        return key;
      }
    }
    return null;
  }

  public List<String> getFilesByTemplate(String jobId, String templateId) {
    JobDTO jobDTO = jobService.selectByPrimaryKey(jobId);
    CheckParaUtils.checkDbObjectValidity(jobDTO, logger, String.format(DcsErrorCode.JOB_NULL.getDesc(), jobId));

    List<FileFilterRuleDto> fileFilterRuleList = JSONUtils.toList(jobDTO.getSelectedFile(), FileFilterRuleDto.class);
    FileFilterRuleDto fileFilter = null;
    if (CollectionUtils.isNotEmpty(fileFilterRuleList)) {
      for(FileFilterRuleDto fileFilterRule : fileFilterRuleList){
        if(fileFilterRule.getTemplateId().equals(templateId)){
          fileFilter = fileFilterRule;
          break;
        }
      }
    }

    List<String> fileList = new ArrayList<>();
    if(fileFilter != null){
      List<CatalogBean> catalogBeanList = fileFilter.getCatalogBeanList();
      if (CollectionUtils.isNotEmpty(catalogBeanList)) {
        fileList = catalogBeanList.stream().map(catalogBean -> catalogBean.getCatalogPath()).collect(Collectors.toList());
      }
    }
    return fileList;
  }

}